package com.rainsoft.center.isec.common.utils;

import com.alibaba.fastjson.annotation.JSONField;
import com.rainsoft.center.isec.common.annotation.HColum;
import com.rainsoft.center.isec.common.entity.Constants;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.lang.reflect.Field;
import java.sql.Timestamp;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;


public class ISecHBaseUtils {
	private static String zkhost = Constants.ZK_HOSTS;
	private static String hbasePort = "2181";
	private static Configuration conf;
	private static Connection conn;
	private static Admin admin;
	private final static Logger logger = LoggerFactory.getLogger(ISecHBaseUtils.class);

	static {
		conf = HBaseConfiguration.create();
		conf.set("hbase.zookeeper.quorum", zkhost);
		conf.set("hbase.zookeeper.property.clientPort", hbasePort);
		try {
			conn = ConnectionFactory.createConnection(conf);
			admin = conn.getAdmin();
		} catch (IOException e) {
			e.printStackTrace();
		}
	}

	/**
	 * 创建表,多列簇
	 *
	 * @param tableName 表名
	 * @param families  列族列表
	 * @throws IOException
	 */
	public static void createTable(String tableName, String[] families) throws IOException {
		TableName tName = TableName.valueOf(tableName);
		if (admin.tableExists(tName)) {
			logger.warn(tableName + " exists.");
		} else {
			HTableDescriptor hTableDesc = new HTableDescriptor(tName);
			for (String col : families) {
				HColumnDescriptor hColumnDesc = new HColumnDescriptor(col);
				hTableDesc.addFamily(hColumnDesc);
			}
			admin.createTable(hTableDesc);
		}
	}

	/**
	 * 创建表,单列簇
	 *
	 * @param tableName 表名
	 * @param family    列族
	 * @throws IOException
	 */
	public static void createTable(String tableName, String family) throws IOException {
		TableName tName = TableName.valueOf(tableName);
		if (admin.tableExists(tName)) {
			logger.warn(tableName + " exists.");
		} else {
			HTableDescriptor hTableDesc = new HTableDescriptor(tName);
			HColumnDescriptor hColumnDesc = new HColumnDescriptor(family);
			hTableDesc.addFamily(hColumnDesc);
			admin.createTable(hTableDesc);
		}
	}


	/**
	 * 传入对象列表添加或者更新多条数据
	 *
	 * @param tableName 表名
	 * @param rowKey    key
	 * @param family    列簇名
	 * @param iterable  对象列表
	 */
	public static void addOrUpdateBatchData(String tableName,String rowKey,String family, List<?> iterable) {
		try {

			Table table = conn.getTable(TableName.valueOf(tableName));
			List<Put> putList = new ArrayList<>();
			for (Object t : iterable) {
				Put put = generatePut(t, rowKey, family);
				putList.add(put);
			}
			table.put(putList);
			table.close();
		} catch (IOException e) {
			e.printStackTrace();
			logger.warn("表:{}添加数据失败", tableName);
		}
	}








	/**
	 * 添加或者跟新一条数据
	 *
	 * @param tableName 表名
	 * @param rowKey    key
	 * @param family    列簇名
	 * @param column    列名
	 * @param value     值
	 */
	public static void addOrUpdateData(String tableName, String rowKey, String family, String column, String value) {
		try {
			Table table = conn.getTable(TableName.valueOf(tableName));
			Put put = new Put(Bytes.toBytes(rowKey));
			put.addColumn(Bytes.toBytes(family), Bytes.toBytes(column), Bytes.toBytes(value));
			table.put(put);
			table.close();
		} catch (IOException e) {
			e.printStackTrace();
			logger.warn("表:{}添加失败", tableName);
		}
	}

	/**
	 * 通过rowkey获取数据
	 *
	 * @param tableName
	 * @param rowkey
	 * @return
	 */
	public static Result getResult(String tableName, String rowkey) {
		Result result = null;
		try {
			Table table = conn.getTable(TableName.valueOf(tableName));
			Get get = new Get(Bytes.toBytes(rowkey));
			result = table.get(get);
		} catch (IOException e) {
			e.printStackTrace();
		}
		return result;
	}

	/**
	 * 查询某指定列的值
	 *
	 * @param tableName
	 * @param rowKey
	 * @param familyName
	 * @param column
	 * @return
	 */
	public static Result getResultByColumn(String tableName, String rowKey, String familyName, String column) {
		Result result = null;
		try {
			Table table = conn.getTable(TableName.valueOf(tableName));
			Get get = new Get(Bytes.toBytes(rowKey));
			get.addColumn(Bytes.toBytes(familyName), Bytes.toBytes(column));
			result = table.get(get);
		} catch (IOException e) {
			e.printStackTrace();
		}
		return result;

	}

	/**
	 * 遍历表（可以按条件过滤查找）
	 *
	 * @param tableName
	 * @param start_rowkey
	 * @param stop_rowkey
	 * @throws IOException
	 */
	public static void getResultScann(String tableName, String start_rowkey,
	                                  String stop_rowkey) throws IOException {
		Scan scan = new Scan();
		if (start_rowkey != null && !"".equals(start_rowkey)) {
			scan.setStartRow(Bytes.toBytes(start_rowkey));
		}
		if (stop_rowkey != null && !"".equals(stop_rowkey)) {
			scan.setStopRow(Bytes.toBytes(stop_rowkey));
		}
		ResultScanner rs = null;
		Table table = conn.getTable(TableName.valueOf(tableName));
		try {
			rs = table.getScanner(scan);
			for (Result r : rs) {
				for (Cell kv : r.listCells()) {
					System.out.println("row:" + Bytes.toString(kv.getRowArray(), kv.getRowOffset(), kv.getRowLength()));
					System.out.println("family:" + Bytes.toString(kv.getFamilyArray(), kv.getFamilyOffset(), kv.getFamilyLength()));
					System.out.println("qualifier:" + Bytes.toString(kv.getQualifierArray(), kv.getQualifierOffset(), kv.getQualifierLength()));
					System.out.println("value:" + Bytes.toString(kv.getValueArray(), kv.getValueOffset(), kv.getValueLength()));
					System.out.println("timestamp:" + kv.getTimestamp());
					System.out.println("-------------------------------------------");
				}
			}
		} finally {
			if (rs != null) {
				rs.close();
			}
		}
	}

	/**
	 * 删除指定的列
	 */
	public static void deleteColumn(String tableName, String rowKey, String familyName, String column) {
		Table table = null;
		try {
			table = conn.getTable(TableName.valueOf(tableName));
			Delete delete = new Delete(Bytes.toBytes(rowKey));
			delete.addColumn(Bytes.toBytes(familyName), Bytes.toBytes(column));
			table.delete(delete);
		} catch (IOException e) {
			e.printStackTrace();
		} finally {
			try {
				if (table != null) {
					table.close();
				}
			} catch (IOException e) {
				e.printStackTrace();
			}
		}
	}

	/**
	 * 删除表
	 *
	 * @param tableName
	 * @throws IOException
	 */
	public static void deleteTable(String tableName) throws IOException {
		admin.disableTable(TableName.valueOf(tableName));
		admin.deleteTable(TableName.valueOf(tableName));
		logger.info(tableName + "is deleted!");
	}


	/**
	 * 根据列条件查询
	 *
	 * @param tableName
	 */
	public static void queryByCondition(String tableName, String familyName, String qualifier, String value) {

		try {
			HTable table = (HTable) conn.getTable(TableName.valueOf(tableName));

			//ValueFilter filter = new ValueFilter( CompareFilter.CompareOp.EQUAL, new BinaryComparator(Bytes.toBytes(value)));
			// Filter filter = new SingleColumnValueFilter(Bytes.toBytes(familyName), Bytes.toBytes(qualifier), CompareFilter.CompareOp.EQUAL, Bytes.toBytes
			// (value)); // 当列familyName的值为value时进行查询
			SingleColumnValueFilter filter = new SingleColumnValueFilter(Bytes.toBytes(familyName), Bytes.toBytes(qualifier), CompareFilter.CompareOp.EQUAL,
					new SubstringComparator(value)); // 当列familyName的值为value时进行查询
			filter.setFilterIfMissing(true);  //true跳过不存在qualifier 的这列  ；false 则是通过这行

			Scan s = new Scan();
			s.setFilter(filter);

			ResultScanner rs = table.getScanner(s);
			for (Result r : rs) {
				System.out.println("获得到rowkey:" + new String(r.getRow()));
				for (Cell keyValue : r.rawCells()) {
					System.out.println("列：" + new String(CellUtil.cloneFamily(keyValue)) + ":" + new String(CellUtil.cloneQualifier(keyValue)) + "====值:" +
							new String(CellUtil.cloneValue(keyValue)));
				}
			}
			rs.close();
		} catch (Exception e) {
			e.printStackTrace();
		}

	}

	/**
	 * 多条件查询
	 *
	 * @param tableName
	 */
	public static void queryByConditions(String tableName, String[] familyNames, String[] qualifiers, String[] values) {

		try {
			HTable table = (HTable) conn.getTable(TableName.valueOf(tableName));
			List<Filter> filters = new ArrayList<Filter>();
			if (familyNames != null && familyNames.length > 0) {
				int i = 0;
				for (String familyName : familyNames) {
					Filter filter = new SingleColumnValueFilter(Bytes.toBytes(familyName), Bytes.toBytes(qualifiers[i]), CompareFilter.CompareOp.EQUAL, Bytes
							.toBytes(values[i]));
					filters.add(filter);
					i++;
				}
			}
			FilterList filterList = new FilterList(filters);
			Scan scan = new Scan();
			scan.setFilter(filterList);
			ResultScanner rs = table.getScanner(scan);
			for (Result r : rs) {
				System.out.println("获得到rowkey:" + new String(r.getRow()));
				for (Cell keyValue : r.rawCells()) {
					System.out.println("列：" + new String(CellUtil.cloneFamily(keyValue)) + ":" + new String(CellUtil.cloneQualifier(keyValue)) + "====值:" +
							new String(CellUtil.cloneValue(keyValue)));
				}
			}
			rs.close();
		} catch (Exception e) {
			e.printStackTrace();
		}
	}


	/**
	 * 根据列条件模糊查询（like）
	 *
	 * @param tableName
	 */
	public static void queryLikeCondition(String tableName, String familyName, String qualifier, String regexStr) {

		try {
			HTable table = (HTable) conn.getTable(TableName.valueOf(tableName));
			Filter filter = new SingleColumnValueFilter(Bytes.toBytes(familyName), Bytes.toBytes(qualifier), CompareFilter.CompareOp.EQUAL, new
					RegexStringComparator(regexStr)); // 当列familyName的值为value时进行查询
			Scan s = new Scan();
			s.setFilter(filter);
			ResultScanner rs = table.getScanner(s);
			for (Result r : rs) {
				System.out.println("获得到rowkey:" + new String(r.getRow()));
				for (Cell keyValue : r.rawCells()) {
					System.out.println("列：" + new String(CellUtil.cloneFamily(keyValue)) + ":" + new String(CellUtil.cloneQualifier(keyValue)) + "====值:" +
							new String(CellUtil.cloneValue(keyValue)));
				}
			}
			rs.close();
		} catch (Exception e) {
			e.printStackTrace();
		}

	}

	/**
	 * 通过传入的对象生成Put
	 *
	 * @param o      对象
	 * @param row    表的rowkey
	 * @param family 表的列簇名
	 * @return
	 * @throws IllegalAccessException
	 */
	public static Put generatePut(Object o, String row, String family) {
		if (row == null) {
			return null;
		}
		Class<?> clazz = o.getClass();
		Put put = new Put(Bytes.toBytes(row));
		List<Field> fields = new ArrayList<>();
		fields.addAll(Arrays.asList(clazz.getDeclaredFields()));
		Class<?> superClazz = clazz;
		while ((superClazz = superClazz.getSuperclass()) != null) {
			Field[] lst = superClazz.getDeclaredFields();
			for (Field field : lst) {
				if (!contains(fields, field)) {
					fields.add(field);
				}
			}
		}

		for (Field declaredField : fields) {
			String name = declaredField.getName();
			if (name.equals("serialVersionUID")) {
				continue;
			}
			HColum hColum = declaredField.getAnnotation(HColum.class);
			if (hColum != null) {
				boolean ignore = hColum.ignore();
				if (ignore) {
					continue;
				}
			}
			JSONField jsonField = declaredField.getAnnotation(JSONField.class);
			if (jsonField != null) {
				String jsonName = jsonField.name();
				if (!"".equals(jsonName)) {
					name = jsonName;
				}
			}
			declaredField.setAccessible(true);
			try {
				Class<?> type = declaredField.getType();
				Object value = declaredField.get(o);
				if (value != null) {
					if (type == java.sql.Date.class) {
						java.sql.Date date = (java.sql.Date) value;
						value = new SimpleDateFormat("yyyy-MM-dd").format(date);
					} else if (type.isAssignableFrom(java.sql.Timestamp.class)) {
						Timestamp timestamp = (Timestamp) value;
						value = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(timestamp);
					} else if (type == java.util.Date.class) {
						java.util.Date date = (java.util.Date) value;
						value = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(date);
					}
				}
				put.addColumn(Bytes.toBytes(family), Bytes.toBytes(name), Bytes.toBytes(String.valueOf(value)));
			} catch (IllegalAccessException e) {
				e.printStackTrace();
			}
		}
		return put;
	}

	/**
	 * 判断是否包含相同的field
	 *
	 * @param list
	 * @param field
	 * @return
	 */
	protected static boolean contains(List<Field> list, Field field) {
		if (list == null || list.isEmpty()) {
			return false;
		}
		for (Field f : list) {
			if (f.getName().equals(field.getName()) && f.getType() == field.getType()) {
				return true;
			}
		}
		return false;
	}


	/**
	 * 格式化输出
	 *
	 * @param result
	 */
	public static void showCell(Result result) {
		Cell[] cells = result.rawCells();
		for (Cell cell : cells) {
			System.out.println("RowName: " + Arrays.toString(CellUtil.cloneRow(cell)));
			System.out.println("Timetamp: " + cell.getTimestamp());
			System.out.println("column Family: " + Arrays.toString(CellUtil.cloneFamily(cell)));
			System.out.println("row Name: " + Arrays.toString(CellUtil.cloneQualifier(cell)));
			System.out.println("value: " + Arrays.toString(CellUtil.cloneValue(cell)));
		}
	}


}
